
package mvcc

import (
	"fmt"
	"math"

	"go.etcd.io/etcd/api/v3/mvccpb"
	"go.etcd.io/etcd/pkg/v3/adt"
)

var (
	// watchBatchMaxRevs是
	// 一次可以发送给非同步观察者的最大不同版本。声明为
	// var而不是const，用于测试目的。
	watchBatchMaxRevs = 1000
)

// 结构体eventBatch中可以封装多个Event实例，
type eventBatch struct {
	// 其中记录的Event实例是按照revision排序的
	evs []mvccpb.Event
	// 记录当前eventBatch中记录的Event实例来自多少个不同的mainrevision，
	// 特别注意，是不相同的mainrevision值的个数。
	revs int
	// 由于当前eventBatch中记录的Event个数达到上限之后，
	// 后续Event实例无法加入该eventBatch中，
	// 该字段记录了无法加入该eventBatch实例的第一个Event实例对应的mainrevision值。
	moreRev int64
}

// eventBatch.add（）方法实现了将Event实例添加到eventBatch中保存的功能
func (eb *eventBatch) add(ev mvccpb.Event) {
	// 检测revs是否达到上限，如采达到上限，则直接返回
	if eb.revs > watchBatchMaxRevs {
		// 最大批量
		return
	}
	// 第一次添加Event实例
	if len(eb.evs) == 0 {
		eb.revs = 1
		// 将Event实例添加到evs字段中保存
		eb.evs = append(eb.evs, ev)
		return
	}

	// 获取最后一个Event实例对应的mainrevision值
	ebRev := eb.evs[len(eb.evs)-1].Kv.ModRevision
	// 新增Event实例对应的main revision值
	evRev := ev.Kv.ModRevision
	if evRev > ebRev {	// 比较两个main revision位
		eb.revs++	// 遂得revs字段
		if eb.revs > watchBatchMaxRevs {
			// 当前eventBatch实例中记录的Event实例已达上恨，则更新moreRev字段，
			// 记录最后一个无法加入的Event实例的main revision值
			eb.moreRev = evRev
			return
		}
	}
	// 如果能继续添加Event实例，则将Event实例添加到evs中
	eb.evs = append(eb.evs, ev)
}

type watcherBatch map[*watcher]*eventBatch

func (wb watcherBatch) add(w *watcher, ev mvccpb.Event) {
	eb := wb[w]
	if eb == nil {
		eb = &eventBatch{}
		wb[w] = eb
	}
	eb.add(ev)
}

// newWatcherBatch将观察者映射到其匹配的事件。它可以让观察者快速查找
// 事件。
func newWatcherBatch(wg *watcherGroup, evs []mvccpb.Event) watcherBatch {
	if len(wg.watchers) == 0 {
		return nil
	}

	wb := make(watcherBatch)
	for _, ev := range evs {
		for w := range wg.watcherSetByKey(string(ev.Kv.Key)) {
			if ev.Kv.ModRevision >= w.minRev {
				// 不要重复通知
				wb.add(w, ev)
			}
		}
	}
	return wb
}

type watcherSet map[*watcher]struct{}

func (w watcherSet) add(wa *watcher) {
	if _, ok := w[wa]; ok {
		panic("add watcher twice!")
	}
	w[wa] = struct{}{}
}

func (w watcherSet) union(ws watcherSet) {
	for wa := range ws {
		w.add(wa)
	}
}

func (w watcherSet) delete(wa *watcher) {
	if _, ok := w[wa]; !ok {
		panic("removing missing watcher!")
	}
	delete(w, wa)
}

type watcherSetByKey map[string]watcherSet

func (w watcherSetByKey) add(wa *watcher) {
	set := w[string(wa.key)]
	if set == nil {
		set = make(watcherSet)
		w[string(wa.key)] = set
	}
	set.add(wa)
}

func (w watcherSetByKey) delete(wa *watcher) bool {
	k := string(wa.key)
	if v, ok := w[k]; ok {
		if _, ok := v[wa]; ok {
			delete(v, wa)
			if len(v) == 0 {
				// 删除集合；没有留下任何东西
				delete(w, k)
			}
			return true
		}
	}
	return false
}

// watcherGroup是由其范围组织的观察者的集合
type watcherGroup struct {
	// 该字段记录监昕单个Key的watcher实例，watcherSetByKey实际上就是map[string]watcherSet类型，
	// 该map中的key就是监听的原始Key值。watcherSetByK巳y提供了add（）和delet巳（）两个方法，
	// 分别用来添加和删除指定的watcher实例，
	keyWatchers watcherSetByKey
	// 该字段记录进行范围监听的watcher实例，IntervalTree （线段树〉是二叉树的一种变形，
	// 线段树将一个区间划分成一些单元区间，每一个区间对应线段树的一个叶节点。
	// 假设在线段树中一个非叶子节点［a,b］，那么它的左儿子节点表示的区间范围为是［a,(a+b)/2］，
	// 右儿子节点表示的区间范围为［（a+b)/2+1.b］。
	ranges adt.IntervalTree
	// 该字段中记录了当前watcherGroup实例中全部的watcher实例，
	// 这里的watcherSet实际上就是map[*watcher]struct{}类型。
	// 在watcherSet中定义了add（）、delete（）和union（）三个方法，
	// 分别用来添加、删除watcher实例和合并其他watcherSet实例，
	watchers watcherSet
}

func newWatcherGroup() watcherGroup {
	return watcherGroup{
		keyWatchers: make(watcherSetByKey),
		ranges:      adt.NewIntervalTree(),
		watchers:    make(watcherSet),
	}
}

// 在向（un) synced watcherGroup中添加watcher实例时，
// 都是通过调用watcherGroup.add()方法实现的
func (wg *watcherGroup) add(wa *watcher) {
	wg.watchers.add(wa)	// 首先将watcher实例添加圭1Jwatchers字段中保存
	if wa.end == nil {	// watcher.end为空表示只监听单个Key，则将其添加到keyWatchers中保存
		wg.keyWatchers.add(wa)
		return
	}

	// 如果待添加的是范围watcher，则执行下面的添加逻辑
	// 根据待添加watcher的key和end字段创建IntervalTree中的一个节点
	ivl := adt.NewStringAffineInterval(string(wa.key), string(wa.end))
	if iv := wg.ranges.Find(ivl); iv != nil {
		// 如果在IntervalTree中查找到了对应节点，则将该watcher添加到对应的watcherSet中
		iv.Val.(watcherSet).add(wa)
		return
	}

	// 当前工ntervalTree中没有对应节点，则创建对应的watcherSet，并添加到IntervalTree中
	ws := make(watcherSet)
	ws.add(wa)
	wg.ranges.Insert(ivl, ws)
}

// 包含的是给定密钥在组中是否有观察者。
func (wg *watcherGroup) contains(key string) bool {
	_, ok := wg.keyWatchers[key]
	return ok || wg.ranges.Intersects(adt.NewStringAffinePoint(key))
}

// size给出组中唯一观察者的数量。
func (wg *watcherGroup) size() int { return len(wg.watchers) }

// 删除从组中删除一个观察者。
func (wg *watcherGroup) delete(wa *watcher) bool {
	if _, ok := wg.watchers[wa]; !ok {
		return false
	}
	wg.watchers.delete(wa)
	if wa.end == nil {
		wg.keyWatchers.delete(wa)
		return true
	}

	ivl := adt.NewStringAffineInterval(string(wa.key), string(wa.end))
	iv := wg.ranges.Find(ivl)
	if iv == nil {
		return false
	}

	ws := iv.Val.(watcherSet)
	delete(ws, wa)
	if len(ws) == 0 {
		// 删除间隔丢失的观察程序
		if ok := wg.ranges.Delete(ivl); !ok {
			panic("could not remove watcher from interval tree")
		}
	}

	return true
}

// watcherGroup.choose（）方法。该方法会根据 unsynced watcherGroup 中记录的 watcher 个数对其进 行分批返回 。
// 另外，它还会获取该批 watcher 实例 中 查找最 小的 minRev 字段。
func (wg *watcherGroup) choose(maxWatchers int, curRev, compactRev int64) (*watcherGroup, int64) {
	// 当前 unsynced watcherGroup 中记录的 watcher 个数未达到指定上限 （默认值为 512 ），
	// 则直接调用 unsynced watcherGroup 的 chooseAll（）方法获取所有未完成同步的 watcher 中最小的 minRev 字段值
	if len(wg.watchers) < maxWatchers {
		return wg, wg.chooseAll(curRev, compactRev)
	}
	// 如果当前 unsynced watcherGroup 中的 watcher 个数超过指定上限，则需要分批处理
	ret := newWatcherGroup()	// 创建新的 watcherGroup 实例
	// 从 unsynced watcherGroup 中遍历得到指定量的 watcher ， 并添加到新建的 watcherGroup 实例中
	for w := range wg.watchers {
		if maxWatchers <= 0 {
			break
		}
		maxWatchers--
		ret.add(w)
	}
	// 依然通过 chooseAll（）方法从该批 watcher 实例中查找最小的 minRev 字段
	return &ret, ret.chooseAll(curRev, compactRev)
}

// watcherGroup.chooseAll()方法，它会遍历 watcherGroup 中记录的全部 watcher 实例，
// 记录其中最小的 minRev 宇段值并返回，
func (wg *watcherGroup) chooseAll(curRev, compactRev int64) int64 {
	minRev := int64(math.MaxInt64)	// 用于记录该 watcherGroup 中最小的 minRev 字段
	for w := range wg.watchers {	// 追历 watcherGroup 中全部 watcher 实例
		if w.minRev > curRev {	//
			if !w.restore {
				panic(fmt.Errorf("watcher minimum revision %d should not exceed current revision %d", w.minRev, curRev))
			}
			w.restore = false
		}
		if w.minRev < compactRev {	// 该 watcher 因为压缩操作而被删除
			select {
			// 创建 watchRespons e （注意，其 CompactRevision 字段是有值的），并写入 watcher . ch 通道
			case w.ch <- WatchResponse{WatchID: w.id, CompactRevision: compactRev}:
				w.compacted = true	// 将 watcher . compacted 设置为 false ，表示因压缩操作而被删除
				wg.delete(w)	// 将该 wa tcher 实例从 watcherGroup 中删除
			default:
				// 如果 watcher.ch 通道阻塞， 贝1J 下次调用 chooseAll （）方法时再尝试重新删除该 watcher 实例
			}
			continue
		}
		if minRev > w.minRev {	// 更新 minRev 遍历，记录最小的 watcher.minRev 字段
			minRev = w.minRev
		}
	}
	return minRev
}

// watcherSetByKey获取接收给定密钥上事件的观察器集。
func (wg *watcherGroup) watcherSetByKey(key string) watcherSet {
	wkeys := wg.keyWatchers[key]
	wranges := wg.ranges.Stab(adt.NewStringAffinePoint(key))

	// 零拷贝案例
	switch {
	case len(wranges) == 0:
		// 无需合并范围或拷贝；重复使用单键集
		return wkeys
	case len(wranges) == 0 && len(wkeys) == 0:
		return nil
	case len(wranges) == 1 && len(wkeys) == 0:
		return wranges[0].Val.(watcherSet)
	}

	// 复制案例
	ret := make(watcherSet)
	ret.union(wg.keyWatchers[key])
	for _, item := range wranges {
		ret.union(item.Val.(watcherSet))
	}
	return ret
}
